package com.reactor.调度器;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class Demo {

	public static void main(String[] args) {
		//使用 create()方法创建一个新的 Flux 对象，其中包含唯一的元素是当前线程的名称。
		Flux.create(sink -> {
		    sink.next(Thread.currentThread().getName());
		    sink.complete();
		})
		//接着是两对 publishOn()和 map()方法，其作用是先切换执行时的调度器，再把当前的线程名称作为前缀添加。
		.publishOn(Schedulers.single())
		.map(x -> String.format("[%s] %s", Thread.currentThread().getName(), x))
		.publishOn(Schedulers.elastic())
		.map(x -> String.format("[%s] %s", Thread.currentThread().getName(), x))
		//最后通过 subscribeOn()方法来改变流产生时的执行方式。
		.subscribeOn(Schedulers.parallel())
		.toStream()
		//运行之后的结果是[elastic-2] [single-1] parallel-1。最内层的线程名字 parallel-1 来自产生流中元素时使用的 Schedulers.parallel()调度器，中间的线程名称 single-1 来自第一个 map 操作之前的 Schedulers.single()调度器，最外层的线程名字 elastic-2 来自第二个 map 操作之前的 Schedulers.elastic()调度器。
		.forEach(System.out::println);

		Flux.fromArray(new String[]{"A", "B", "C", "D"})
        .publishOn(Schedulers.newSingle("TEST-SINGLE", true))
        .map(x -> String.format("[%s]: %s", Thread.currentThread().getName(), x))
        .toStream()
        .forEach(System.out::println);
	}

}
